package prv.liuyao.proxy.queue;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.Executors;
import java.util.function.BiConsumer;

public class DisruptorBiMQ<T, U> implements ProxyBiMQ<T, U> {

    protected Disruptor<LyEvent<T, U>> disruptor;

    public DisruptorBiMQ() {
        this(128);
    }

    public DisruptorBiMQ(int ringBufferSize) {
        this.disruptor = new Disruptor(new EventFactory<LyEvent<T, U>>() {
            @Override
            public LyEvent<T, U> newInstance() {
                return new LyEvent<T, U>();
            }
        }, ringBufferSize, Executors.defaultThreadFactory());
        this.disruptor.setDefaultExceptionHandler(new ExceptionHandler<LyEvent<T, U>>() {
            @Override
            public void handleEventException(Throwable throwable, long l, LyEvent<T, U> tLyEvent) {
                System.out.println(tLyEvent.getT() + ": " + tLyEvent.getU());
                throwable.printStackTrace();
            }
            @Override
            public void handleOnStartException(Throwable throwable) {
                throwable.printStackTrace();
            }
            @Override
            public void handleOnShutdownException(Throwable throwable) {
                throwable.printStackTrace();
            }
        });
    }

    public void push(T t, U u) {
        // 生产者
        RingBuffer<LyEvent<T, U>> ringBuffer = this.disruptor.getRingBuffer();
        long sequence = ringBuffer.next(); // 下一个可用位置
        ringBuffer.get(sequence).setT(t).setU(u);
        ringBuffer.publish(sequence);
    }

    public DisruptorBiMQ<T, U> registryConsumer(BiConsumer<T, U> consumer) {
        this.disruptor.handleEventsWith(new EventHandler<LyEvent<T, U>>() {
            @Override
            public void onEvent(LyEvent<T, U> tLyEvent, long l, boolean b) throws Exception {
                consumer.accept(tLyEvent.getT(), tLyEvent.getU());
            }
        });
        return this;
    }

    public void start() {
        this.disruptor.start();
    }

    public void shutdown() {
        this.disruptor.shutdown();
    }

    private class LyEvent<T, U> {
        private T t;
        private U u;

        public U getU() {
            return u;
        }

        public LyEvent<T, U> setU(U u) {
            this.u = u;
            return this;
        }

        public T getT() {
            return t;
        }

        public LyEvent<T, U> setT(T t) {
            this.t = t;
            return this;
        }
    }
}
